package com.lsx143.wordcount.day8;


import com.lsx143.wordcount.bean.GoodState;
import com.lsx143.wordcount.bean.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Comparator;
import java.util.TreeSet;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/12/18 15:48
 */
public class Flink_TopN_Teacher {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        env
                .readTextFile("input/UserBehavior.csv")
                .map(line -> {
                    String[] data = line.split(",");
                    return new UserBehavior(
                            Long.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2]),
                            data[3],
                            Long.parseLong(data[4]) * 1000
                    );
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((ub, ts) -> ub.getTimestamp())
                )
                .filter(ub -> "pv".equals(ub.getBehavior()))
                .keyBy(UserBehavior::getItemId)  // 按照商品id分组, 统计每个商品的点击量
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(30)))
                .aggregate(
                        new AggregateFunction<UserBehavior, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(UserBehavior value, Long acc) {
                                return acc + 1;
                            }

                            @Override
                            public Long getResult(Long acc) {
                                return acc;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Long, GoodState, Long, TimeWindow>() {
                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<Long> elements,
                                                Collector<GoodState> out) throws Exception {
                                Long count = elements.iterator().next();
                                long wEnd = ctx.window().getEnd();
                                out.collect(new GoodState(key, wEnd, count));
                            }
                        }
                )
                .keyBy(GoodState::getWEnd)
                .process(new KeyedProcessFunction<Long, GoodState, String>() {

                    private ValueState<TreeSet<GoodState>> treeSetState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // 使用list状态存储所有值
                        treeSetState = getRuntimeContext()
                                .getState(new ValueStateDescriptor<TreeSet<GoodState>>("treeSetState", TypeInformation.of(new TypeHint<TreeSet<GoodState>>() {
                                })));
                    }

                    @Override
                    public void processElement(GoodState GoodState,
                                               Context ctx,
                                               Collector<String> out) throws Exception {
                        // 第一个来了之后, 应该注册一个定时器
                        TreeSet<GoodState> set = treeSetState.value();
                        if (set == null) {
                            set = new TreeSet<>();
                            treeSetState.update(set);

                            ctx.timerService().registerEventTimeTimer(GoodState.getWEnd() + 2000);
                        }

                        set.add(GoodState);

                        // 当set集合长度等于4, 则删除最后一个
                        if (set.size() == 4) {
                            set.pollLast();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<String> out) throws Exception {
                        TreeSet<GoodState> set = treeSetState.value();

                        StringBuilder msg = new StringBuilder("-----------\n");
                        for (GoodState GoodState : set) {
                            msg.append(GoodState).append("\n");
                        }

                        out.collect(msg.toString());

                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
